package com.maowei.parser;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.maowei.common.CanalConstants;
import com.maowei.common.GlobalConstant;
import com.maowei.event.EventInfo;
import com.maowei.util.StringUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * rowData数据 自定义解析 - 默认实现类
 *
 * Tips : 一般会给数据库增加非业务字段 create_by(创建人) create_time(创建时间) update_by(修改人) update_time(修改时间) delete_flag(删除状态)
 *          一般某些业务下的删除可能会是状态删除 . 如 : 修改delete_flag字段为1(已删除的)
 *
 * @author maowei
 * @date 2021/8/24
 */
public class DefaultRowDtaCustomParser implements RowDataCustomParser{

    @Override
    public JSONObject parse(EventInfo eventInfo) {
        if (eventInfo == null){
            return null;
        }
        CanalEntry.RowData rowData = eventInfo.getRowData();
        if (rowData == null){
            return null;
        }
        CanalEntry.EventType eventType = eventInfo.getEventType();
        if (CanalEntry.EventType.INSERT == eventType){
            //新增
            //保留新增后的数据
            Map<String, String> afterColumnMap = flat2Map(rowData.getAfterColumnsList());
            return addOriginalColumn(doParse(eventInfo,afterColumnMap),null,afterColumnMap);
        }else if (CanalEntry.EventType.UPDATE == eventType){
            //修改
            //保留前后的数据, 如是软删除, 则强制为删除类型, 并保留删除前的数据
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            Map<String, String> afterColumnMap = flat2Map(afterColumnsList);
            String deleteFlag = afterColumnMap.get(GlobalConstant.DELETE_FLAG_COLUMN);
            if (GlobalConstant.DELETED_FLAG == Integer.parseInt(deleteFlag)){
                //更新为删除状态, 软删除, 则强制为删除类型
                Map<String, String> beforeColumnMap = flat2Map(rowData.getBeforeColumnsList());
                return addOriginalColumn(doParse(eventInfo,beforeColumnMap,CanalEntry.EventType.DELETE),beforeColumnMap,null);
            }
            //实际更新的字段
            Map<String, String> updatedColumn = flat2Map(afterColumnsList.stream()
                    .filter(column -> column.getUpdated())
                    .collect(Collectors.toList()));
            if (StringUtils.isEmpty(updatedColumn)){
                //无更新数据, 无需处理
                return null;
            }
            Map<String, String> beforeColumnMap = flat2Map(rowData.getBeforeColumnsList());
            //变化的数据
            Map<String,Map<String,String>> changedData = new HashMap<>(updatedColumn.size());
            for (Map.Entry<String, String> nameValueEntry : updatedColumn.entrySet()) {
                String columnName = nameValueEntry.getKey();
                Map<String,String> beforeAfterMap = new HashMap<>(2);
                beforeAfterMap.put(CanalConstants.Parser.CHANGED_COLUMN_BEFORE,beforeColumnMap.get(columnName));
                beforeAfterMap.put(CanalConstants.Parser.CHANGED_COLUMN_AFTER,nameValueEntry.getValue());
                changedData.put(columnName,beforeAfterMap);
            }
            return addOriginalColumn(doParse(eventInfo,changedData),beforeColumnMap,afterColumnMap);
        }else if (CanalEntry.EventType.DELETE == eventType){
            //删除
            //保留删除前的数据
            Map<String, String> beforeColumnMap = flat2Map(rowData.getBeforeColumnsList());
            return addOriginalColumn(doParse(eventInfo,beforeColumnMap),beforeColumnMap,null);
        }
        //TODO 后期可扩展其他事件类型
        return null;
    }

    private JSONObject doParse(EventInfo eventInfo,Object columnData){
        return doParse(eventInfo,columnData,null);
    }

    /**
     * 执行解析包装数据
     * @param eventInfo
     * @param columnData
     * @param overrideEventTyp 强制覆盖事件类型(如: 软删除, 实际为更新操作, 业务上为删除)
     * @return
     */
    private JSONObject doParse(EventInfo eventInfo,Object columnData,CanalEntry.EventType overrideEventTyp){
        if (columnData == null){
            return null;
        }
        JSONObject parsedJSON = new JSONObject();
        parsedJSON.put(CanalConstants.Parser.SCHEMA_NAME,eventInfo.getTableInfo().getSchemaName());
        parsedJSON.put(CanalConstants.Parser.TABLE_NAME,eventInfo.getTableInfo().getTableName());
        if (overrideEventTyp == null){
            parsedJSON.put(CanalConstants.Parser.EVNENT_TYPE,eventInfo.getEventType().name().toLowerCase());
        }else{
            parsedJSON.put(CanalConstants.Parser.EVNENT_TYPE,overrideEventTyp.name().toLowerCase());
        }
        parsedJSON.put(CanalConstants.Parser.CHANGED_COLUMN,columnData);
        return parsedJSON;
    }

    /**
     * 增加上原始的数据(操作前后的数据)
     * @param parsedJSON
     * @param beforeColumnMap
     * @param afterColumnMap
     */
    private JSONObject addOriginalColumn(JSONObject parsedJSON,Map<String, String> beforeColumnMap,Map<String, String> afterColumnMap){
        if (parsedJSON == null){
            return null;
        }
        if (StringUtils.isNotEmpty(beforeColumnMap)){
            parsedJSON.put(CanalConstants.Parser.BEFORE_COLUMN,beforeColumnMap);
        }
        if (StringUtils.isNotEmpty(afterColumnMap)){
            parsedJSON.put(CanalConstants.Parser.AFTER_COLUMN,afterColumnMap);
        }
        return parsedJSON;
    }


    /**
     * 平铺成map
     * @param columns
     * @return
     */
    private Map<String,String> flat2Map(List<CanalEntry.Column> columns){
        if (StringUtils.isEmpty(columns)){
            return null;
        }
        return columns.stream().collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
    }

}
